package io.wen.bd.s6m2.spark

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

// ip排序，依次按4个部分比较
case class Ip(ip: String) extends Ordered[Ip] {
  override def compare(that: Ip): Int = {
    val a = this.ip.split("\\.")
    val b = that.ip.split("\\.")
    if (a(0).toInt.compareTo(b(0).toInt) != 0) return a(0).toInt.compareTo(b(0).toInt)
    if (a(1).toInt.compareTo(b(1).toInt) != 0) return a(1).toInt.compareTo(b(1).toInt)
    if (a(2).toInt.compareTo(b(2).toInt) != 0) return a(2).toInt.compareTo(b(2).toInt)
    if (a(3).toInt.compareTo(b(3).toInt) != 0) return a(3).toInt.compareTo(b(3).toInt)
    0
  }
}

case class HttpData(timestamp: String, ip: Ip, website: String, uri: String, agent: String, startIp: Ip)

case class IpData(startIp: Ip, endIp: Ip, city: String)

/**
  * 将 http.log 文件中的 ip 转换为地址。如将 122.228.96.111 转为温州，并统计各城市的总访问量
  */
object IP2LocationTransformer {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .appName("spark session")
      .master("local[*]")
      .getOrCreate()

    val sc = spark.sparkContext
    sc.setLogLevel("WARN")
    import spark.implicits._

    val httpDataPath = "file:///D:\\workspace\\project\\bd-hw\\stage6-module2-spark-core-sql\\data\\http.log"
    val ipDataPath = "file:///D:\\workspace\\project\\bd-hw\\stage6-module2-spark-core-sql\\data\\ip.dat"

    // 读取ip数据，解析为IpData
    val ipRDD = sc
      .textFile(ipDataPath)
      .map(line => line.split("\\|"))
      .filter(_.length == 15)
      .map(arr => IpData(Ip(arr(0)),Ip(arr(1)),arr(7)))

    // 将startIp列排序广播，用于http数据拓展startIp列
    val startIpArr = ipRDD.map(item => item.startIp).collect().sorted
    val startIpArrBC = sc.broadcast(startIpArr)

    // 读取http数据解析为 HttpData, startIp赋默认值
    val httpRDD = sc
      .textFile(httpDataPath)
      .map(line => line.split("\\|"))
      .filter(_.length == 5)
      .map(arr => HttpData(arr(0), Ip(arr(1)), arr(2), arr(3), arr(4), Ip("0.0.0.0")))

    // http数据扩展startIp列，二分查找广播startIp数据
    val extHttpRDD = httpRDD.map(item => {
      val arr = startIpArrBC.value
      val itemStartIp = findStartIp(arr, item.ip)
      HttpData(item.timestamp, item.ip, item.website, item.uri, item.agent, itemStartIp)
    })

    val extHttpDS = extHttpRDD.toDS()
    val ipDS = ipRDD.toDS()

    // map join，将添加startIp列后的ip数据广播
    val resultDF = extHttpDS.join(broadcast(ipDS), extHttpDS("startIp") === ipDS("startIp"))
    resultDF.show(5)

    // 统计各城市的总访问量
    resultDF.groupBy("city").count.show
  }


  /**
    * 二分查找获取startIp
    */
  def findStartIp(ipArr: Array[Ip], ip: Ip): Ip = {
    var lo = 1
    var hi = ipArr.length

    if (ip > ipArr(hi - 1)) return ipArr(hi - 1)

    while (lo <= hi) {
      val mid = lo + (hi - lo) / 2
      if (ip < ipArr(mid)){
        if (ip >= ipArr(mid - 1)) {
          return ipArr(mid - 1)
        } else {
          hi = mid - 1
        }
      } else {
        if (ip < ipArr(mid - 1)) {
          return ipArr(mid - 1)
        } else {
          lo = mid + 1
        }
      }
    }

    Ip("0.0.0.0")
  }
}